[GSoC 2026] Kafka Streams runner: read unbounded sources - #39611

The runner could only read sources that finish, which is the wrong shape for
what it is: a Kafka Streams application is a long-running stream processor,
and bounded data has more efficient homes.

Bounded and unbounded reads share a URN and are distinguished by the payload,
so ReadTranslator branches on it. The unbounded processor polls its reader on
a wall-clock punctuator rather than draining it once, since advance()
returning false means nothing is available right now rather than that the
source is finished, and takes at most maxBundleSize elements per turn so a
fast source cannot starve the rest of the topology. Its watermark comes from
UnboundedReader#getWatermark() instead of jumping to the end of time when the
input runs out, which is what lets downstream windows close on a stream that
never finishes.

The reader's checkpoint mark is written to a persistent state store after the
elements it covers have been forwarded, so it can never claim more progress
than was emitted, and the reader is created from the stored mark so a restart
resumes where it left off. finalizeCheckpoint is not called yet: a mark
should only be finalized once durably committed, which needs a pre-commit
hook the runner does not have. The source is also read by a single reader, so
splits are not distributed across instances.

UnboundedReadTest drives a genuinely unbounded CountingSource and asserts
more than one poll's worth of elements arrive, contiguously from zero, which
is what separates a polled source from a drained one.

…ore reading

Reuses ReadTranslation.boundedSourceFromProto directly rather than wrapping
it, with one try/catch covering parsing and hydration for both kinds of
source.

Polls in batches that run back to back rather than returning after one, since
waiting for the next punctuation after every batch capped throughput at a
batch per interval. The run is bounded all the same: a source that always has
data would otherwise never let the poll return and the Kafka Streams thread
would never get back to committing or to the rest of the topology, so at most
readCheckpointNumBundles batches are taken before yielding. Polling also
stops entirely, and the punctuator is cancelled, once the reader's watermark
reaches the end of time, which is the source saying it will produce nothing
further.

Adds --readCheckpointNumBundles, since taking a checkpoint mark can be costly
and is not worth doing on every batch. The cost of a larger value is that
more elements are replayed after a restart, because the reader resumes from
the last mark stored.

Asks the source to split before creating a reader. A source is not obliged to
be readable in its unsplit form, and split() is where several of them do
their setup, so going through it even for a single reader is the supported
path.

Adds a test that a source reaching the end of time stops being polled and
delivers each element exactly once.

Splitting was being done inside the processor, which is the wrong place: it
would run once per task instance rather than once for the pipeline, and the
contract does not define splitting a source that has already been split. It
now happens in ReadTranslator, and the processor is handed a source that is
ready to read.

The count passed to split() is only a hint, so what comes back is checked.
Taking the first of several splits and ignoring the rest would quietly drop
whatever those parts would have produced, which is data loss rather than a
missing feature, so translation fails with an explanation instead. Reading
several splits in parallel is still not supported.

UnboundedReadTest covers it with a source that returns two splits whatever it
is asked for.